package tailf

import (
	"context"
	"go_kafka_log/common"
	"go_kafka_log/kafka"

	"github.com/hpcloud/tail"
)

type tailTask struct {
	path   string
	topic  string
	tObj   *tail.Tail
	ctx    context.Context
	cancel context.CancelFunc
}

//var (
//	confChan chan []common.CollectEntry
//)

// 读取日志，发送至 kafka
func (t *tailTask) run() {

	for {
		select {
		// 如果 此 goroutine 监控的日志文件已经取消，则退出此 goroutine
		case <-t.ctx.Done():
			// 退出 goroutine
			return
		// 从 tail channel 获取信息
		case line := <-t.tObj.Lines:
			// 如果是空行，略过
			if len(line.Text) == 0 {
				continue
			}
			//topic := viper.GetString("collect.topic")
			println("发现新的消息！！！！！！！！！！！！！！！", t.path, t.topic)

			// 将日志的最新信息 打入 kafka包的 MsgChan,而不是直接发送给 kafka对应的数据的分区
			kafka.SendToKafka(t.topic, line.Text)

		}

	}

}

func SendNewConf(conf []common.CollectEntry) {
	ttMgr.confChan <- conf
}
